package com.breeze.toolkit.rxandroid;

import java.util.concurrent.TimeUnit;

import rx.Observable;
import rx.Observer;
import rx.Subscriber;
import rx.Subscription;
import rx.android.schedulers.AndroidSchedulers;
import rx.schedulers.Schedulers;

/**
 * Created by john on 2016/8/23.
 */
public class RxAsyncEvent<T> implements Observable.OnSubscribe<T>, Observer<T> {
	private RxAsyncInterface<T> mListener;
	private Subscription mSubscription;

	public RxAsyncEvent(RxAsyncInterface<T> t) {
		mListener = t;
	}

	public void subscribe() {
		if (mListener != null) {
			mListener.onPreExecute();
		}
		mSubscription = Observable.create(this)
				.subscribeOn(Schedulers.io())
				.observeOn(AndroidSchedulers.mainThread())
				.subscribe(this);
	}

	public void delaySubscribeMilliseconds(int delay) {
		if (mListener != null) {
			mListener.onPreExecute();
		}
		delaySubscribe(delay, TimeUnit.MILLISECONDS);
	}

	public void delaySubscribe(int delay, TimeUnit unit) {
		mSubscription = Observable.create(this)
				.subscribeOn(Schedulers.io())
				.observeOn(AndroidSchedulers.mainThread())
				.delaySubscription(delay, unit)
				.subscribe(this);
	}

	public void unsubscribe() {
		if (mSubscription != null && !mSubscription.isUnsubscribed()) {
			mSubscription.unsubscribe();
		}
	}

	@Override
	public void call(Subscriber<? super T> subscriber) {
		if (mListener != null && !subscriber.isUnsubscribed()) {
			subscriber.onNext(mListener.doInBackground());
		}
	}

	@Override
	public void onCompleted() {

	}

	@Override
	public void onError(Throwable e) {

	}

	@Override
	public void onNext(T t) {
		if (mListener != null) {
			mListener.onPostExecute(t);
		}
	}
}